package utils;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import java.util.List;

public class KafkaHelper {

    public static String[] getKafkaFields(String jsonArr) {
        List<String> dataKeys = JSONArray.parseArray(jsonArr, String.class);
        String[] kafkaFields = new String[dataKeys.size()];
        for (int i = 0, len = dataKeys.size(); i < len; i++) {
            kafkaFields[i] = dataKeys.get(i);
        }
        return  kafkaFields;
    }

    public static String[] getKafkaFields(List<String> dataKeys) {
        String[] kafkaFields = new String[dataKeys.size()];
        for (int i = 0, len = dataKeys.size(); i < len; i++) {
            kafkaFields[i] = dataKeys.get(i);
        }
        return  kafkaFields;
    }

    public static TypeInformation<?>[] getKafkaFieldsTypeInfo(String jsonArr) {
        List<String> dataTypes = JSONArray.parseArray(jsonArr, String.class);
        TypeInformation<?>[] kafkaTypes = new TypeInformation[dataTypes.size()];
        for (int i = 0, len = dataTypes.size(); i < len; i++) {
            kafkaTypes[i] = FlinkDataType.getTypeInformation(dataTypes.get(i));
        }
        return  kafkaTypes;
    }

    public static TypeInformation<?>[] getKafkaFieldsTypeInfo(List<String> dataTypes) {
        TypeInformation<?>[] kafkaTypes = new TypeInformation[dataTypes.size()];
        for (int i = 0, len = dataTypes.size(); i < len; i++) {
            kafkaTypes[i] = FlinkDataType.getTypeInformation(dataTypes.get(i));
        }
        return  kafkaTypes;
    }

}
